package com.atguigu.actual.edu0417.app.dwd.log;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.actual.edu0417.app.BaseApp;
import com.atguigu.actual.edu0417.common.EduConfig;
import com.atguigu.actual.edu0417.unils.DateFormatUtil;
import com.atguigu.actual.edu0417.unils.MyKafkaUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SideOutputDataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @author: 洛尘
 * @since: 2023-10-20 08:52
 * @description: 流量域
 **/
public class DwdTrafficBaseLogSplit extends BaseApp {

    public static void main(String[] args) {
        DwdTrafficBaseLogSplit dwdTrafficBaseLogSplit = new DwdTrafficBaseLogSplit();
        dwdTrafficBaseLogSplit.base(args);
    }

    @Override
    public void process(StreamExecutionEnvironment env) {
        //从topic_log读取数据，声明消费者组并创建消费者对象
        String groupId="dwd_traffic_log_group";

        KafkaSource<String> kafkaSource = MyKafkaUtil.getKafkaSource(EduConfig.KAFKA_TOPIC_LOG, groupId);

        SingleOutputStreamOperator<String> kafkaDS = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka_log_source");
//        kafkaDS.print(">>>");
            //对读取的数据进行类型转化并进行ETL处理，将脏数据放入侧输出流
        OutputTag<String> dirtyTag = new OutputTag<String>("dirty_tag"){};

        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String jsonStr, Context ctx, Collector<JSONObject> out) throws Exception {
                        try {
                            JSONObject jsonObj = JSON.parseObject(jsonStr);
                            out.collect(jsonObj);
                        } catch (Exception e) {
                            e.printStackTrace();
                            ctx.output(dirtyTag, jsonStr);
                        }

                    }
                }
        );
        SideOutputDataStream<String> dirtyDS = jsonObjDS.getSideOutput(dirtyTag);

        //将脏数据放入kafka主题中
        KafkaSink<String> kafkaSink = MyKafkaUtil.getKafkaSink("dirty_date");
        dirtyDS.sinkTo(kafkaSink);

        //使用flink的状态编程，修复修老访客标记
        //按照mid分组
        KeyedStream<JSONObject, String> keyedDS = jsonObjDS.keyBy(jsonObj -> jsonObj.getJSONObject("common").getString("mid"));

        //修复
        SingleOutputStreamOperator<JSONObject> fixedDS = keyedDS.map(
                new RichMapFunction<JSONObject, JSONObject>() {
                    //不能在声明状态的时候直接初始化，因为此时获取不到运行上下文
                    private ValueState<String> lastValueState;

                    @Override
                    public void open(Configuration parameters) throws Exception {

                        ValueStateDescriptor<String> lastValueDescriptor = new ValueStateDescriptor<>("lastValueDescriptor", String.class);
                        lastValueState = getRuntimeContext().getState(lastValueDescriptor);
                    }

                    @Override
                    public JSONObject map(JSONObject jsonObj) throws Exception {
                        //获取新老访客的标记的值
                        String isNew = jsonObj.getJSONObject("common").getString("is_new");
                        //从状态中获取当前设备的首次访问日期
                        String lastValueDate = lastValueState.value();
                        //获取当前日期对应时间毫秒
                        Long ts = jsonObj.getLong("ts");
                        String curDate = DateFormatUtil.toDate(ts);

                        if ("1".equals(isNew)) {
                            if (StringUtils.isEmpty(lastValueDate)) {
                                //如果监控状态为空的话，那么证明事首次访问app
                                lastValueState.update(curDate);
                            } else {
                                //如果监控状态不是null，且首次访问日期不是当日，说明访问的事老访客，将isNew改成0
                                if (!lastValueDate.equals(curDate)) {
                                    isNew = "0";
                                    jsonObj.getJSONObject("common").put("is_new", isNew);
                                }
                            }
                        } else {
                            //如果is_new 为0
                            //如果监控状态为null，说明访问app的是老访客，但是本次是该访客的页面日志首次进入程序中
                            //当前端新老访客状态标记丢失时，日志进入程序被判定为新访客，Flink程序就可以纠正被误判的访客状态标记
                            //只要将状态中的日期设置为今天之前即可，本程序选择将状态改为昨天
                            if (StringUtils.isEmpty(lastValueDate)) {
                                String yesterDay = DateFormatUtil.toDate(ts - 24 * 3600 * 1000);
                                lastValueState.update(yesterDay);
                            }
                        }
                        return jsonObj;
                    }
                }
        );
//        fixedDS.print("!!!");
        //对不同的日志进行分流
        //错误日志-错误侧输出流 启动日志-启动侧输出流 曝光日志-曝光侧输出流 动作日志-动作侧输出流 页面日志-主流
        OutputTag<String> errTag = new OutputTag<String>("err_tag"){};
        OutputTag<String> actionTag = new OutputTag<String>("action_tag"){};
        OutputTag<String> displayTag = new OutputTag<String>("display_tag"){};
        OutputTag<String> startTag = new OutputTag<String>("start_tag"){};
        OutputTag<String> appVideoTag = new OutputTag<String>("appVideo_tag"){};

        //分流
        SingleOutputStreamOperator<String> pageDS = fixedDS.process(
                new ProcessFunction<JSONObject, String>() {
                    @Override
                    public void processElement(JSONObject jsonObj, Context ctx, Collector<String> out) throws Exception {
                        //处理错误日志
                        JSONObject errObj = jsonObj.getJSONObject("err");
                        if (errObj != null) {
                            //将错误日志输出到错误侧输出流
                            ctx.output(errTag, jsonObj.toJSONString());
                            jsonObj.remove("err");
                        }
                        //处理启动日志
                        JSONObject startObj = jsonObj.getJSONObject("start");
                        JSONObject appVideoObj = jsonObj.getJSONObject("appVideo");
                        if (startObj != null) {
                            ctx.output(startTag, jsonObj.toJSONString());
                            //处理视频日志
                        }else if (appVideoObj!=null){
                            ctx.output(appVideoTag,jsonObj.toJSONString());
                        }else {
                            //页面日志和曝光日志还有动作日志
                            JSONObject pageObj = jsonObj.getJSONObject("page");
                            Long ts = jsonObj.getLong("ts");
                            JSONObject commonObj = jsonObj.getJSONObject("common");

                            //处理曝光日志
                            JSONArray displayObjArr = jsonObj.getJSONArray("displays");
                            if (displayObjArr != null && displayObjArr.size() > 0) {
                                for (int i = 0; i < displayObjArr.size(); i++) {
                                    JSONObject displayObjArrJSONObject = displayObjArr.getJSONObject(i);
                                    //定义一个新的json对象用来封装display日志
                                    JSONObject newJsonObject = new JSONObject();
                                    newJsonObject.put("common", commonObj);
                                    newJsonObject.put("display", displayObjArrJSONObject);
                                    newJsonObject.put("page", pageObj);
                                    newJsonObject.put("ts", ts);
                                    //将曝光日志输出到侧输出流
                                    ctx.output(displayTag, newJsonObject.toJSONString());
                                }
                                jsonObj.remove("displays");

                            }
                            //处理动作日志
                            JSONArray actionArr = jsonObj.getJSONArray("actions");
                            if (actionArr != null && actionArr.size() > 0) {
                                for (int i = 0; i < actionArr.size(); i++) {
                                    JSONObject arrJSONObject = actionArr.getJSONObject(i);
                                    //定义一个新的json对象来封装action日志
                                    JSONObject newjsonObject = new JSONObject();
                                    newjsonObject.put("common", commonObj);
                                    newjsonObject.put("action", arrJSONObject);
                                    newjsonObject.put("page", pageObj);

                                    ctx.output(actionTag, newjsonObject.toJSONString());
                                }
                                jsonObj.remove("actions");
                            }
                            out.collect(jsonObj.toJSONString());
                        }
                    }
                }
        );
        //将不同的流输出到不同的kafka主题
        SideOutputDataStream<String> errDS = pageDS.getSideOutput(errTag);
        SideOutputDataStream<String> startDS = pageDS.getSideOutput(startTag);
        SideOutputDataStream<String> actionDS = pageDS.getSideOutput(actionTag);
        SideOutputDataStream<String> displayDS = pageDS.getSideOutput(displayTag);
        SideOutputDataStream<String> appVideoDS = pageDS.getSideOutput(appVideoTag);

        pageDS.print("page>>>");
        errDS.print("err>>>");
        startDS.print("start>>>");
        actionDS.print("action>>>");
        displayDS.print("display>>>");
        appVideoDS.print("appVideo>>>");

        pageDS.sinkTo(MyKafkaUtil.getKafkaSink("dwd_traffic_page_log"));
        startDS.sinkTo(MyKafkaUtil.getKafkaSink("dwd_traffic_start_log"));
        errDS.sinkTo(MyKafkaUtil.getKafkaSink("dwd_traffic_err_log"));
        actionDS.sinkTo(MyKafkaUtil.getKafkaSink("dwd_traffic_action_log"));
        displayDS.sinkTo(MyKafkaUtil.getKafkaSink("dwd_traffic_display_log"));
        appVideoDS.sinkTo(MyKafkaUtil.getKafkaSink("dwd_traffic_appVideo_log"));

    }
}